package com.xiashitech.agent.diagnosis.server;

import com.xiashitech.agent.diagnosis.advisor.AdviceListener;
import com.xiashitech.agent.diagnosis.advisor.AdviceWeaver;
import com.xiashitech.agent.diagnosis.advisor.Enhancer;
import com.xiashitech.agent.diagnosis.advisor.InvokeTraceable;
import com.xiashitech.agent.diagnosis.command.Command;
import com.xiashitech.agent.diagnosis.command.Command.Action;
import com.xiashitech.agent.diagnosis.command.Command.GetEnhancerAction;
import com.xiashitech.agent.diagnosis.command.Command.Printer;
import com.xiashitech.agent.diagnosis.command.QuitCommand;
import com.xiashitech.agent.diagnosis.command.ShutdownCommand;
import com.xiashitech.agent.diagnosis.exception.GaExecuteException;
import com.xiashitech.agent.diagnosis.util.DiagnosisHttpUtil;
import com.xiashitech.agent.diagnosis.util.affect.Affect;
import com.xiashitech.agent.diagnosis.util.affect.EnhancerAffect;
import com.xiashitech.agent.diagnosis.util.affect.RowAffect;
import org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.lang.instrument.Instrumentation;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;

import static com.xiashitech.agent.diagnosis.util.GaStringUtils.ABORT_MSG;
import static com.xiashitech.agent.diagnosis.util.GaStringUtils.getCauseMessage;
import static java.lang.String.format;

/**
 * 命令处理器
 */
public class XSCommandHandler implements CommandHandler {

    protected final Logger logger = Logger.getLogger("XSCommandHandler");

    private final GaServer gaServer;
    private final Instrumentation inst;

    public XSCommandHandler(GaServer gaServer, Instrumentation inst) {
        this.gaServer = gaServer;
        this.inst = inst;
    }

    @Override
    public void executeCommand(final String line, final Command command, final Session session) throws IOException {

        try {
            execute(session, command, line);

            if (command instanceof QuitCommand) {//退出命令，需要关闭会话
                session.destroy();
            } else if (command instanceof ShutdownCommand) {//关闭命令，需要关闭整个服务端
                XSCommandHandler.this.gaServer.destroy();
            }
        } catch (GaExecuteException e) {//命令执行错误
            logger.log(Level.INFO,"command execute failed."+e.getMessage());
            final String cause = getCauseMessage(e);
            if (StringUtils.isNotBlank(cause)) {
                //输出错误信息
                DiagnosisHttpUtil.sendResult(session.getSessionId(),line,"Command execution failed. cause : " + cause);
            } else {
                //输出错误信息
                DiagnosisHttpUtil.sendResult(session.getSessionId(),line,"Command execution failed.\n");
            }
        }

    }


    /*
     * 执行命令
     */
    private void execute(final Session session, final Command command, final String line) throws GaExecuteException, IOException {

        // 是否结束输入引用
        final AtomicBoolean isFinishRef = new AtomicBoolean(false);

        // 消息发送者
        final Printer printer = new Printer() {

            @Override
            public Printer print(boolean isF, String message) {

                if(isFinishRef.get()) {
                    return this;
                }

                final BlockingQueue<String> writeQueue = session.getWriteQueue();
                if (null != message) {
                    if (!writeQueue.offer(message)) {
                        logger.log(Level.WARNING,"offer message failed. write-queue.size() was:" + writeQueue.size());
                    }
                }

                if (isF) {
                    finish();
                }

                return this;

            }

            @Override
            public Printer println(boolean isF, String message) {
                return print(isF, message + "\n");
            }

            @Override
            public Printer print(String message) {
                return print(false, message);
            }

            @Override
            public Printer println(String message) {
                return println(false, message);
            }

            @Override
            public void finish() {
                isFinishRef.set(true);
            }

        };


        try {

            // 影响反馈
            final Affect affect;
            final Action action = command.getAction();

            // 无任何后续动作的动作
            if (action instanceof Command.SilentAction) {
                affect = new Affect();
                ((Command.SilentAction) action).action(session, inst, printer);
            }

            // 需要反馈行影响范围的动作
            else if (action instanceof Command.RowAction) {
                affect = new RowAffect();
                final RowAffect rowAffect = ((Command.RowAction) action).action(session, inst, printer);
                ((RowAffect) affect).rCnt(rowAffect.rCnt());
            }

            // 需要做类增强的动作
            else if (action instanceof GetEnhancerAction) {

                affect = new EnhancerAffect();

                // 执行命令动作 & 获取增强器
                final Command.GetEnhancer getEnhancer = ((GetEnhancerAction) action).action(session, inst, printer);
                final int lock = session.getLock();
                final AdviceListener listener = getEnhancer.getAdviceListener();
                final EnhancerAffect enhancerAffect = Enhancer.enhance(
                        inst,
                        lock,
                        listener instanceof InvokeTraceable,
                        getEnhancer.getPointCut()
                );

                // 这里做个补偿,如果在enhance期间,unLock被调用了,则补偿性放弃
                if (session.getLock() == lock) {
                    // 注册通知监听器
                    AdviceWeaver.reg(lock, listener);

                    if (!session.isSilent()) {
                        printer.println(ABORT_MSG);
                    }

                    ((EnhancerAffect) affect).cCnt(enhancerAffect.cCnt());
                    ((EnhancerAffect) affect).mCnt(enhancerAffect.mCnt());
                    ((EnhancerAffect) affect).getClassDumpFiles().addAll(enhancerAffect.getClassDumpFiles());
                }
            }

            // 其他自定义动作
            else {
                // do nothing...
                affect = new Affect();
            }

            if (!session.isSilent()) {
                // 记录下命令执行的执行信息
                printer.print(false, affect.toString() + "\n");
            }

        }

        // 命令执行错误必须纪录
        catch (Throwable t) {
            throw new GaExecuteException(format("execute failed. sessionId=%s", session.getSessionId()), t);
        }

        // 跑任务
        jobRunning(line, session, isFinishRef);

    }

    private void jobRunning(String line, Session session, AtomicBoolean isFinishRef) throws IOException, GaExecuteException {

        final Thread currentThread = Thread.currentThread();
        final BlockingQueue<String> writeQueue = session.getWriteQueue();
        while (GaServer.getIsDebugForDiagnosis() && !session.isDestroy()
                && !currentThread.isInterrupted()
                && session.isLocked()) {

            try {
                final String segment = writeQueue.poll(200, TimeUnit.MILLISECONDS);
                if (segment == null && isFinishRef.get()) {
                    session.unLock();
                    break;
                }
                if(segment != null) {
                    //touch the session
                    session.touch();
                    DiagnosisHttpUtil.sendResult(session.getSessionId(),line,segment);
                }
            } catch (InterruptedException e) {
                currentThread.interrupt();
            }

        }//while command running
    }

    /*
     * 输出到网络
     */
    private void write(SocketChannel socketChannel, String message, Charset charset) throws IOException {
        write(socketChannel, charset.encode(message));
    }

    private void write(SocketChannel socketChannel, ByteBuffer buffer) throws IOException {
        while (buffer.hasRemaining()
                && socketChannel.isConnected()) {
            if (-1 == socketChannel.write(buffer)) {
                // socket broken
                throw new IOException("write EOF");
            }
        }
    }

}
